﻿using DotNetCore.CAP;
using DotNetCore.CAP.Internal;
using DotNetCore.CAP.Messages;
using DotNetCore.CAP.Transport;
using Microsoft.Extensions.Logging;
using MQTTnet;
#if !NET8_0_OR_GREATER
using MQTTnet.Client;
using MQTTnet.Client.Publishing;
#endif
using MQTTnet.Protocol;
using System;
using System.Threading.Tasks;
using Newtonsoft.Json;
using Microsoft.Extensions.Options;
using System.Linq;

namespace DotNetCore.CAP.MQTT
{
    public class MqttTransport : ITransport
    {

        private readonly IConnectionChannelPool _connectionPool;
        private readonly ILogger _logger;
        readonly bool _useString;
        readonly MqttQualityOfServiceLevel _qos;

        public MqttTransport(ILogger<MqttTransport> logger, IOptions<CAPMqttOptions> mqttOptions, IConnectionChannelPool connectionPool)
        {
            _useString = mqttOptions.Value.UseString;
            _qos = mqttOptions.Value.Qos;
            _logger = logger;
            _connectionPool = connectionPool;
        }
        public BrokerAddress BrokerAddress => new BrokerAddress("Mqtt", _connectionPool.HostAddress);
     
        public static object BytesToObject(byte[] bytes, Type type)
        {
            string jsonString = System.Text.Encoding.UTF8.GetString(bytes);
            return JsonConvert.DeserializeObject(jsonString, type);
        }
        object GetSendPack(TransportMessage message)
        {
            if (_useString)
            {
                var mqttMessage = new MqttMessage();
                mqttMessage.Headers = message.Headers;
                mqttMessage.Body = BytesToObject(message.Body.ToArray(), typeof(object));
                return mqttMessage;
            }
            else
            {
                var mqttMessage = new MqttMessageBytes();
                mqttMessage.Headers = message.Headers;
                mqttMessage.Body = message.Body.ToArray();
                return mqttMessage;
            }
        }    
        public async Task<OperateResult> SendAsync(TransportMessage message)
        {
            var mqttClient = _connectionPool.Rent();

            try
            {
                var bytes = JsonConvert.SerializeObject(GetSendPack(message));
                var applicationMessageBuild = new MqttApplicationMessageBuilder()
                .WithTopic(message.GetName())
                .WithPayload(bytes)
                .WithQualityOfServiceLevel(_qos);
                var result = await mqttClient.PublishAsync(applicationMessageBuild.Build());

                if (result.ReasonCode == MqttClientPublishReasonCode.Success)
                {
                    _logger.LogDebug($"mqtt topic message [{message.GetName()}] has been published.");

                    return OperateResult.Success;
                }
                throw new PublisherSentFailedException("mqtt message persisted failed!");
            }
            catch (Exception ex)
            {
                var wapperEx = new PublisherSentFailedException(ex.Message, ex);

                return OperateResult.Failed(wapperEx);
            }
            finally
            {
                _connectionPool.Return(mqttClient);
            }
        }
    }
}
